package com.angelasp.main;

import com.angelasp.domain.PgConnection;
import com.angelasp.domain.PgProperties;
import com.ververica.cdc.connectors.postgres.PostgreSQLSource;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Properties;


/**
 * @Author yangb
 * @Date 2022/08/13 20:56
 * @Version 1.0
 */

public class PgSQLSourceExample {

    public static void main(String[] args) throws Exception {
        Properties properties =new Properties();
        //这里设置全量还是增量该属性是 never不执行初始一致性快照，但是会同步后续数据库的更改记录 默认是全量+增量 initial
        //properties.setProperty("snapshot.mode","never");
        properties.setProperty("snapshot.mode","initial");
        //获取配置文件信息
        PgConnection pgConnection = PgProperties.getConfigProperties();
        SourceFunction<String> sourceFunction = PostgreSQLSource.<String>builder()
                .hostname(pgConnection.getHostName())
                .port(pgConnection.getPort())
                .database(pgConnection.getDataBase())
                .schemaList(pgConnection.getSchemaList())
                .tableList(pgConnection.getTableList())
                .username(pgConnection.getUsername())
                .password(pgConnection.getPassword())
                .slotName(pgConnection.getSlotName())
                .decodingPluginName("pgoutput")
                .debeziumProperties(properties)
                .deserializer(new CustomDeserialization())
                // .startupOptions(StartupOptions.initial())
                .build();

        Configuration config = new Configuration();
        //启动时设置 还原点
        if (!StringUtils.isEmpty(pgConnection.getSavepointPath())) {
            String lasCheckpointPath = pgConnection.getSavepointPath();
            config.setString("execution.savepoint.path", lasCheckpointPath);
        }
        // 启动还原点end
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
        env.enableCheckpointing(3000);

        //checkpoint 存放路径官方建议使用hdfs 目前是采用本地存储方式
        env.setStateBackend(new FsStateBackend(pgConnection.getCheckpointDataUrl()));

        env.addSource(sourceFunction)
                .setParallelism(1)
                .addSink(new CustomSink()).setParallelism(1);

        env.execute("flinkPgCDC");


    }
}